package com.qjc.mq.producer;

import com.alibaba.fastjson.JSONObject;
import com.qjc.mq.listener.SendCallbackListener;
import com.qjc.mq.model.OrderStep;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * @ClassName: Producer
 * @Description:
 * @Author: qjc
 * @Date: 2021/12/13 2:18 下午
 */
@Component
@Slf4j
public class Producer {

    @Value(("${rocketmq.producer.topic}"))
    private String myTopicDestination;
    @Value(("${rocketmq.producer.topic}:${rocketmq.producer.sync-tag}"))
    private String syncMsgDestination;
    @Value(("${rocketmq.producer.topic}:${rocketmq.producer.async-tag}"))
    private String asyncMsgDestination;
    @Value(("${rocketmq.producer.topic}:${rocketmq.producer.oneway-tag}"))
    private String oneWayMsgDestination;
    @Value(("${rocketmq.producer.topic}:${rocketmq.producer.sequence-tag}"))
    private String sequenceMsgDestination;
    @Value(("${rocketmq.producer.topic}:${rocketmq.producer.delay-tag}"))
    private String delayMsgDestination;
    @Value(("${rocketmq.producer.topic}:${rocketmq.producer.batch-tag}"))
    private String batchMsgDestination;
    @Value(("${rocketmq.producer.topic}:${rocketmq.producer.transaction-tag}"))
    private String transactionMsgDestination;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void push(Integer i) {
        String msg = "普通消息-" + i;
        rocketMQTemplate.convertAndSend(myTopicDestination, msg);
        System.out.println("普通消息【" + msg + "】发送成功");
    }

    /**
     * 发送同步消息
     */
    public void pushSyncMessage(int i) {
        // 构建消息
        String msg = "同步消息-" + i;
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.KEYS, i)
                .build();
        // 设置发送地和消息信息并发送同步消息
        SendResult sendResult = rocketMQTemplate.syncSend(syncMsgDestination, message);
        // 解析发送结果
        System.out.println("消息内容：" + msg + "，发送结果：" + sendResult.getSendStatus());
    }

    /**
     * 发送异步消息
     */
    public void pushAsyncMessage(int i) {
        // 构建消息
        String msg = "异步消息-" + i;
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.KEYS, i)
                .build();
        // 设置发送地和消息信息并发送异步消息
        rocketMQTemplate.asyncSend(asyncMsgDestination, message, new SendCallbackListener());
    }

    /**
     * 发送单向消息（不关注发送结果：主要用来记录日志）
     * Oneway方式只发送请求不等待应答，即将数据写入客户端的Socket缓冲区就返回，不等待对方返回结果。
     */
    public void pushOneWayMessage(int id) {
        // 构建消息
        String msg = "单向消息-" + id;
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.KEYS, id)
                .build();
        rocketMQTemplate.sendOneWay(oneWayMsgDestination, message);
        System.out.println("单项消息【" + msg + "】发送成功");
    }

    /**
     * 发送包含顺序的单向消息
     * PS：消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了三条消息分别
     * 是订单创建、订单付款、订单完成。消费时要按照这个顺序消费才能有意义，但是同时订单之间是可以
     * 并行消费的。RocketMQ可以严格的保证消息有序。
     */
    public void pushSequenceMessage(int id) {
        // 获取当前订单的操作步骤列表
        List<OrderStep> orderSteps = OrderStep.buildOrderSteps(String.valueOf(id));
        for (OrderStep orderStep : orderSteps) {
            // 构建消息
            String messageStr = "顺序消息-订单id：" + orderStep.getId() + "，操作步骤：" + orderStep.getDesc();
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, orderStep.getId())
                    .build();
            // 设置顺序下发
            rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
                /**
                 * 设置放入同一个队列的规则
                 * @param list 消息列表
                 * @param message 当前消息
                 * @param id 比较的关键信息
                 * @return 消息队列
                 */
                @Override
                public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object id) {
                    // 根据当前消息的id，使用固定算法获取需要下发的队列
                    // （使用当前id和消息队列个数进行取模获取需要下发的队列，id和队列数量一样时，选择的队列肯定一样）
                    int queueNum = Integer.valueOf(String.valueOf(id)) % list.size();
                    log.info(String.format("队列号【%s】，消息内容【%s】", queueNum, new String(message.getBody())));
                    return list.get(queueNum);
                }
            });
            // 设置发送地和消息信息并发送消息（Orderly）
            rocketMQTemplate.syncSendOrderly(sequenceMsgDestination, message, orderStep.getId());
        }
    }

    /**
     * 定时消息（延迟队列）是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。
     * 延迟消息：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 延迟级别:1-18，大于18的按照2h延迟，0代表非延迟消息
     */
    public void pushDelayMessage(int id) {
        String msg = "延迟消息-" + id;
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.KEYS, id)
                .build();
        // 设置消息延迟5s后被消费
        SendResult sendResult = rocketMQTemplate.syncSend(delayMsgDestination, message, 1_000, 2);
        System.out.println("延迟消息内容【" + msg + "】，发送结果：" + sendResult.getSendStatus());
    }

    /**
     * 批量发送消息
     */
    public void pushBatchMessage(int id) {
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            String myId = id + "" + i;
            String msg = "批量消息-" + myId;
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader(RocketMQHeaders.KEYS, myId)
                    .build();
            messages.add(message);
        }
        SendResult sendResult = rocketMQTemplate.syncSend(batchMsgDestination, messages);
        List<String> msgList = new ArrayList<>();
        for (Message message : messages) {
            String msg = message.getPayload().toString();
            msgList.add(msg);
        }
        System.out.println("批量消息内容【" + msgList + "】，发送结果：" + sendResult.getSendStatus());
    }

    /**
     * sql过滤消息
     */
    public void pushSqlMessage(int id) {
        int money = new Random().nextInt(100);
        String type = money % 2 == 0 ? "a" : "b";
        String msg = "sql过滤消息-" + id + "，money=" + money + "，type=" + type;
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.KEYS, id)
                .setHeader("money", money)
                .setHeader("type", type)
                .build();
        SendResult sendResult = rocketMQTemplate.syncSend(myTopicDestination, message);
        System.out.println("sql过滤消息内容【" + msg + "】，发送结果：" + sendResult.getSendStatus());
    }

    /**
     * 事务消息
     */
    public void pushTransactionMessage(int id) {
        String msg = "事务消息-" + id;
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.KEYS, id)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, id)
                .build();
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(transactionMsgDestination, message, null);
        System.out.println("事务消息内容【" + msg + "】，发送结果：" + transactionSendResult.getSendStatus());
    }

}
